package commMqtt

import (
	"errors"
	"gitee.com/fierce_wolf/go-fox-edge-common/commLogger"
	"gitee.com/fierce_wolf/go-fox-edge-common/commUtil/Json"
	"gitee.com/fierce_wolf/go-fox-edge-common/commUtil/Map"
	"gitee.com/fierce_wolf/go-fox-edge-common/commUtil/Method"
	"gitee.com/fierce_wolf/go-fox-edge-common/commUtil/Pair"
	"gitee.com/fierce_wolf/go-fox-edge-common/commUtil/String"
	"github.com/eclipse/paho.mqtt.golang"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"time"
)

type MqttClient struct {
	defaultOpts map[string]interface{} // 缺省的配置参数
	options     *mqtt.ClientOptions    // 工作参数
	client      mqtt.Client            // mqtt客户端
	clientId    string                 // clientId
	topics      sync.Map               // 准备订阅的topic
	lostTime    int64                  // 丢失连接的时间点
	lostDone    chan int               // 丢失时的chan
}

func (e *MqttClient) initDefaultOption() {
	e.defaultOpts = make(map[string]interface{})

	jsn := `
	{
	"host": "broker.emqx.io",
	"name": "mqtt-client",
	"port": 1883,
	"enabled": true,
	"timeout": 5,
	"use-ssl": false,
	"version": "mqtt_3_1_1",
	"clientId": "",
	"password": "",
	"reconnect": true,
	"user-name": "",
	"re-interval": 5000,
	"clean-session": false,
	"keep-alive-secs": 60,
	"buffer-allocator": "heap",
	"read-buffer-size": "8KB",
	"max-bytes-in-message": "10MB"
	}
	`
	e.defaultOpts = Json.BuildMapByJson(jsn)
}

func (e *MqttClient) Disconnect() error {
	if e.client != nil && e.client.IsConnected() {
		// 取消订阅
		e.topics.Range(func(topic, pair interface{}) bool {
			e.client.Unsubscribe(topic.(string))
			return true
		})

		// 断开连接
		for e.client.IsConnected() {
			e.client.Disconnect(250)
			time.Sleep(1 * time.Second)
		}
	}

	return nil
}

func (e *MqttClient) HasTopic(topic string) bool {
	_, ok := e.topics.Load(topic)
	return ok
}

func (e *MqttClient) Topics() []string {
	topics := make([]string, 0)
	e.topics.Range(func(topic, pair interface{}) bool {
		topics = append(topics, topic.(string))
		return true
	})

	return topics
}

func (e *MqttClient) IsConnected() bool {
	return e.client != nil && e.client.IsConnected()
}

func (e *MqttClient) Connect() error {
	if e.client != nil && e.client.IsConnected() {
		// 取消订阅
		e.topics.Range(func(topic, pair interface{}) bool {
			e.client.Unsubscribe(topic.(string))
			return true
		})

		// 断开连接
		for e.client.IsConnected() {
			e.client.Disconnect(250)
			time.Sleep(1 * time.Second)
		}
	}

	commLogger.Info("MQTT:[clientId=" + e.clientId + "] " + "建立连接")

	//  分配一个client实例
	if e.client == nil {
		e.client = mqtt.NewClient(e.options)
	}

	// 重新建立连接
	if token := e.client.Connect(); token.Wait() && token.Error() != nil {
		commLogger.Info("MQTT:[clientId=" + e.clientId + "] " + "连接失败-" + token.Error().Error())
		return token.Error()
	}

	return nil
}

func (e *MqttClient) Subscribe(topic string, handler mqtt.MessageHandler) {
	e.topics.Store(topic, Pair.Pair{Key: handler, Value: false})
}

func (e *MqttClient) subscribe(topic string, handler mqtt.MessageHandler) error {
	if e.client == nil || !e.client.IsConnected() {
		return errors.New("缺失参数：尚未建立连接" + "；clientId=" + e.clientId)
	}

	// 剔除空格
	topic = strings.TrimSpace(topic)

	commLogger.Info("MQTT:[clientId=" + e.clientId + "] " + "开始订阅 " + topic)

	// 订阅topic
	if token := e.client.Subscribe(topic, 1, handler); token.Wait() && token.Error() != nil {
		commLogger.Error("MQTT:[clientId=" + e.clientId + "] " + "订阅失败 " + topic + " 原因：" + token.Error().Error())
		return token.Error()
	}

	// 记录订阅状态
	e.topics.Store(topic, Pair.Pair{Key: handler, Value: true})

	commLogger.Info("MQTT:[clientId=" + e.clientId + "] " + "订阅成功 " + topic)

	return nil
}

func (e *MqttClient) Unsubscribe(topic string) (mqtt.Token, error) {
	if e.client == nil || !e.client.IsConnected() {
		return nil, errors.New("缺失参数：尚未建立连接" + "；clientId=" + e.clientId)
	}

	token := e.client.Unsubscribe(topic)
	commLogger.Info("MQTT:[clientId=" + e.clientId + "] " + "订阅注销-" + topic)
	return token, nil
}

func (e *MqttClient) Unsubscribes(topics ...string) (mqtt.Token, error) {
	if e.client == nil || !e.client.IsConnected() {
		return nil, errors.New("MQTT:[clientId=" + e.clientId + "] " + "缺失参数：尚未建立连接")
	}
	if len(topics) == 0 {
		return nil, errors.New("MQTT:[clientId=" + e.clientId + "] " + "没有订阅")
	}

	token := e.client.Unsubscribe(topics...)
	commLogger.Info("MQTT:[clientId=" + e.clientId + "] " + "订阅注销-" + String.ToString(topics))
	return token, nil
}

func (e *MqttClient) Publish(topic string, payload interface{}, timeout int64) bool {
	token := e.client.Publish(topic, 1, false, payload)
	return token.WaitTimeout(time.Duration(timeout) * time.Millisecond)
}

func (e *MqttClient) GetClient() mqtt.Client {
	return e.client
}

func (e *MqttClient) SetOption(config map[string]interface{}) error {
	for key, value := range e.defaultOpts {
		_, ok := config[key]
		if !ok {
			config[key] = value
		}
	}

	host := Map.GetString(config, "host", "")
	port := Map.GetInt(config, "port", 1883)
	userName := Map.GetString(config, "user-name", "")
	password := Map.GetString(config, "password", "")
	keepAlive := Map.GetInt64(config, "keep-alive-secs", 60)
	reInterval := Map.GetInt64(config, "re-interval", 5000)

	if Method.HasEmpty(host) || !Map.Has(config, "port") {
		return errors.New("缺失参数：host, port不能为空")
	}

	broker := "tcp://" + host + ":" + strconv.Itoa(port)
	commLogger.Info("MQTT:[clientId=" + e.clientId + "]" + " broker=" + broker)

	opts := mqtt.NewClientOptions().AddBroker(broker)

	opts.SetClientID(e.clientId)
	opts.Username = userName
	opts.Password = password

	// 断开自动重连的配置
	opts.ConnectRetry = true                                              // 重试连接
	opts.SetKeepAlive(time.Duration(keepAlive) * time.Second)             // 心跳
	opts.AutoReconnect = true                                             // 自动重连
	opts.SetMaxReconnectInterval(time.Duration(reInterval) * time.Second) // 重连间隔
	opts.CleanSession = false                                             // 不清除session，保证重新连接后可继续执行
	opts.OnConnect = e.connectHandler                                     // 连接后处理
	opts.OnConnectionLost = e.connectLostHandler                          // 断连后处理

	e.options = opts
	return nil
}

func (e *MqttClient) connectHandler(client mqtt.Client) {
	commLogger.Info("MQTT:[clientId=" + e.clientId + "]" + " 连接成功")

	topics := e.Topics()
	for _, topic := range topics {
		pair, _ := e.topics.Load(topic)

		// 重新订阅
		if pair != nil && pair.(Pair.Pair).Key != nil {
			e.subscribe(topic, pair.(Pair.Pair).Key.(mqtt.MessageHandler))
		}
	}

	if e.lostTime == 0 {
		e.lostTime = 0
		return
	}
	e.lostDone <- 1
}

func (e *MqttClient) connectLostHandler(client mqtt.Client, err error) {
	commLogger.Info("MQTT:[clientId=" + e.clientId + "]" + " 连接断开->" + err.Error())

	e.lostTime = time.Now().Unix()
	e.lostDone = make(chan int)
	go e.connectLostFunc(e.lostDone)
}

func (e *MqttClient) connectLostFunc(ch chan int) {
	for {
		select {
		case <-ch:
			runtime.Goexit()
			break
		default:
		}
		time.Sleep(2 * time.Second)
	}
}
